package com.example.rabbitmq.example.rabbitmq.topic;

import com.rabbitmq.client.*;

import java.io.IOException;

public class ShanghaiErrorConsumer {

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new  ConnectionFactory();
        //%2f 是虚拟主机 /
        factory.setUri("amqp://admin:admin@192.168.1.3:5672/%2f");
        final Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();
        String queueName = channel.queueDeclare().getQueue();
        channel.exchangeDeclare("ex.topic", "topic", true, false, null);
        channel.queueBind(queueName, "ex.topic", "shanghai.error.#");

        channel.basicQos(1000);
        channel.basicConsume(queueName,  new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                System.out.println(consumerTag);
                System.out.println(new String(message.getBody()));
            }
        }, consumerTag -> { });

        channel.basicConsume(queueName, false, "consumerTag", new DefaultConsumer(channel) {

            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body)
                    throws IOException{
                //消息的标签, 多个消息还是一个消息
                channel.basicAck(envelope.getDeliveryTag(), false);
                // 是否重新入列， 重发
                channel.basicNack(envelope.getDeliveryTag(), false, true);
                channel.basicReject(envelope.getDeliveryTag(), true);
            }
        });
    }
}
